package core

import (
	"canal-go/global"
	"database/sql"
	"fmt"
	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
	_ "github.com/go-sql-driver/mysql"
	"log"
	"strings"
)

type Canal struct {
	event canal.EventHandler
}

func NewCanal(eventHandle canal.EventHandler) *Canal {
	return &Canal{
		eventHandle,
	}
}

func (cl *Canal) Start() {
	dbConfig := global.APPLICATION.Mysql
	cfg := canal.NewDefaultConfig()
	cfg.Addr = fmt.Sprintf("%s:%d", dbConfig.Host, dbConfig.Port)
	cfg.User = dbConfig.UserName
	cfg.Password = dbConfig.Password
	cfg.Flavor = "mysql"

	canalConfig := global.APPLICATION.Canal
	cfg.Dump.TableDB = canalConfig.Action
	cfg.Dump.Tables = strings.Split(canalConfig.Table, ",")
	cfg.Dump.ExecutionPath = ""
	c, err := canal.NewCanal(cfg)
	if err != nil {
		log.Fatalf("Create core instance error: %v", err)
	}
	c.SetEventHandler(cl.event)

	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%d)/", dbConfig.UserName, dbConfig.Password, dbConfig.Host, dbConfig.Port))
	if err != nil {
		log.Fatalf("Open core instance error: %v", err)
	}
	pos, err := GetCurrentBinlogPosition(db)
	if err != nil {
		log.Fatalf("Get current binlog position error: %v", err)
	}
	log.Println("server id :", cfg.ServerID)
	err = c.RunFrom(pos)
	if err != nil {
		log.Fatalf("Run core instance error: %v", err)
	}
}

func GetCurrentBinlogPosition(db *sql.DB) (mysql.Position, error) {
	defer db.Close()
	var name string
	var position uint32
	var binlogDoDB, binlogIgnoreDB, executedGtidSet sql.NullString
	err := db.QueryRow("SHOW MASTER STATUS").Scan(&name, &position, &binlogDoDB, &binlogIgnoreDB, &executedGtidSet)
	if err != nil {
		return mysql.Position{}, err
	}
	return mysql.Position{Name: name, Pos: position}, nil
}
